package phases

import (
	"context"
	"strings"
	"time"

	"github.com/pkg/errors"
	confv1beta1 "gopkg.openfuyao.cn/bkecommon/cluster/api/v1beta1"
	bkenode "gopkg.openfuyao.cn/bkecommon/cluster/node"
	"k8s.io/apimachinery/pkg/util/wait"
	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
	"sigs.k8s.io/cluster-api/util/conditions"
	ctrl "sigs.k8s.io/controller-runtime"

	bkev1beta1 "gopkg.openfuyao.cn/cluster-api-provider-bke/api/v1beta1"
	"gopkg.openfuyao.cn/cluster-api-provider-bke/pkg/mergecluster"
	"gopkg.openfuyao.cn/cluster-api-provider-bke/pkg/phaseframe"
	"gopkg.openfuyao.cn/cluster-api-provider-bke/pkg/phaseframe/phaseutil"
	"gopkg.openfuyao.cn/cluster-api-provider-bke/utils/clusterutil"
	"gopkg.openfuyao.cn/cluster-api-provider-bke/utils/constant"
)

const (
	EnsureMasterJoinName confv1beta1.BKEClusterPhase = "EnsureMasterJoin"
)

type EnsureMasterJoin struct {
	phaseframe.BasePhase
	nodesToJoin bkenode.Nodes
}

func NewEnsureMasterJoin(ctx *phaseframe.PhaseContext) phaseframe.Phase {
	base := phaseframe.NewBasePhase(ctx, EnsureMasterJoinName)
	return &EnsureMasterJoin{BasePhase: base}
}

func (e *EnsureMasterJoin) Execute() (_ ctrl.Result, err error) {
	if err = e.reconcileMasterJoin(); err != nil {
		return ctrl.Result{}, err
	}
	return ctrl.Result{}, nil
}

func (e *EnsureMasterJoin) NeedExecute(old *bkev1beta1.BKECluster, new *bkev1beta1.BKECluster) (needExecute bool) {
	if !e.BasePhase.DefaultNeedExecute(old, new) {
		return false
	}

	masterInited := false
	// cluster没有初始化的condition，不需要执行
	if err := e.Ctx.RefreshCtxCluster(); err == nil {
		if conditions.IsTrue(e.Ctx.Cluster, clusterv1.ControlPlaneInitializedCondition) {
			masterInited = true
		}
	}
	nodes := phaseutil.GetNeedJoinMasterNodes(new)

	// 第一种情况：首次创建集群，此时masterInited为false,nodes=1,返回false
	if !masterInited && len(nodes) == 1 {
		return false
	}
	// 第二种情况：集群已经初始化，此时masterInited为true,nodes=0,返回false
	if masterInited && len(nodes) == 0 {
		return false
	}
	// 第三种情况：master没有初始化，此时masterInited为false,nodes=0,返回false
	if !masterInited && len(nodes) == 0 {
		return false
	}

	e.SetStatus(bkev1beta1.PhaseWaiting)
	return true
}

func (e *EnsureMasterJoin) reconcileMasterJoin() error {
	ctx, c, bkeCluster, _, log := e.Ctx.Untie()

	if !bkeCluster.Status.AgentStatus.Ready() {
		log.Error(constant.MasterJoinFailedReason, "Agent is not ready")
		return errors.New("agent is not ready")
	}

	if conditions.IsFalse(e.Ctx.Cluster, clusterv1.ControlPlaneInitializedCondition) {
		log.Warn(constant.MasterJoinFailedReason, "master is not initialized, skip join master nodes process")
		return nil
	}

	nodes := phaseutil.GetNeedJoinMasterNodes(bkeCluster)
	e.nodesToJoin = make(bkenode.Nodes, 0)

	log.Info(constant.MasterJoiningReason, "Start join master nodes process")
	log.Info(constant.MasterJoiningReason, "Check whether the node has been associated with a Machine to avoid duplicate creation")

	nodesInfos := []string{}
	nodesCount := 0
	for _, node := range nodes {
		// 正常应该是找不到关联的machine的
		if _, err := phaseutil.NodeToMachine(ctx, c, bkeCluster, node); err == nil {
			log.Warn(constant.MasterJoinedReason, "Node already exists, skip join node. node: %v", node.Hostname)
			bkeCluster.MarkNodeStateFlag(node.IP, bkev1beta1.NodeBootFlag)
		} else {
			nodesInfos = append(nodesInfos, phaseutil.NodeInfo(node))
			nodesCount++
			e.nodesToJoin = append(e.nodesToJoin, node)
		}
	}

	if nodesCount == 0 {
		log.Info(constant.MasterJoinedReason, "All nodes have been joined, skip join node.")
		if err := mergecluster.SyncStatusUntilComplete(c, bkeCluster); err != nil {
			return err
		}
		return nil
	}

	if clusterutil.IsBocloudCluster(bkeCluster) {
		log.Info(constant.MasterJoiningReason, "bocloud cluster need to distribute kube-proxy kubeconfig")
		if err := phaseutil.DistributeKubeProxyKubeConfig(ctx, c, bkeCluster, e.nodesToJoin, log); err != nil {
			return err
		}
	}

	log.Info(constant.MasterJoiningReason, "%d nodes will join, nodes: %v", nodesCount, strings.Join(nodesInfos, ", "))
	scope, err := phaseutil.GetClusterAPIAssociateObjs(ctx, c, e.Ctx.Cluster)
	if err != nil || scope.KubeadmControlPlane == nil {
		log.Error(constant.MasterJoinFailedReason, "Get cluster-api associate objs failed. err: %v", err)
		// cluster api object error, no need to continue
		return err
	}

	specCopy := scope.KubeadmControlPlane.Spec.DeepCopy()
	currentReplicas := specCopy.Replicas
	// 如果节点加入过程中出现异常，需要将节点数量恢复到加入前的状态
	defer func() {
		if err != nil {
			log.Info(constant.MasterJoinFailedReason, "Scale down KubeadmControlPlane replicas to %d.", currentReplicas)
			scope.KubeadmControlPlane.Spec.Replicas = currentReplicas
			if err = phaseutil.ResumeClusterAPIObj(ctx, c, scope.KubeadmControlPlane); err != nil {
				log.Error(constant.MasterJoinFailedReason, "Back up KubeadmControlPlane replicas failed. err: %v", err)
			}
		}
	}()

	bkeNodes := bkenode.Nodes(bkeCluster.Spec.ClusterConfig.Nodes)
	masterNodes := bkeNodes.Master()

	exceptReplicas := *currentReplicas + int32(nodesCount)
	// 无论如何不能超过bkecluster的master数量
	if exceptReplicas > int32(masterNodes.Length()) {
		exceptReplicas = int32(masterNodes.Length())
	}

	scope.KubeadmControlPlane.Spec.Replicas = &exceptReplicas

	log.Info(constant.MasterJoiningReason, "Scale up KubeadmControlPlane replicas %d to %d", *currentReplicas, exceptReplicas)

	if err = phaseutil.ResumeClusterAPIObj(ctx, c, scope.KubeadmControlPlane); err != nil {
		log.Error(constant.MasterJoinFailedReason, "Scale up KubeadmControlPlane replicas failed. err: %v", err)
		// cluster api object error, no need to continue
		return err
	}

	if err = e.waitMasterJoin(nodesCount); err != nil {
		log.Error(constant.MasterJoinFailedReason, "Wait worker join failed. err: %v", err)
		return err
	}

	return nil

}

func (e *EnsureMasterJoin) waitMasterJoin(nodesCount int) error {
	ctx, c, bkeCluster, _, log := e.Ctx.Untie()
	successJoinNode := map[int]confv1beta1.Node{}

	timeOut, err := phaseutil.GetBootTimeOut(e.Ctx.BKECluster)
	if err != nil {
		log.Warn(constant.MasterNotInitReason, "Get boot timeout failed. err: %v", err)
	}

	waitTime := time.Duration(nodesCount) * timeOut
	ctxTimeout, cancel := context.WithTimeout(ctx, waitTime)
	defer cancel()

	pollCount := 0
	err = wait.PollImmediateUntil(1*time.Second, func() (done bool, err error) {
		pollCount++
		for i, node := range e.nodesToJoin {
			if _, ok := successJoinNode[i]; ok {
				continue
			}
			machine, err := phaseutil.NodeToMachine(ctx, c, bkeCluster, node)
			if err != nil {
				continue
			}
			if machine.Status.NodeRef != nil {
				log.Info(constant.MasterJoinSucceedReason, "Master node join success. node: %v", phaseutil.NodeInfo(node))
				successJoinNode[i] = node
			}
		}
		if len(successJoinNode) != len(e.nodesToJoin) {
			if pollCount%10 == 0 {
				log.Info(constant.MasterJoiningReason, "Wait master join. success: %d, total: %d", len(successJoinNode), len(e.nodesToJoin))
			}
			return false, nil
		}

		log.Info(constant.MasterJoiningReason, "Wait master join. success: %d, total: %d", len(successJoinNode), len(e.nodesToJoin))
		return true, nil
	}, ctxTimeout.Done())

	if errors.Is(err, wait.ErrWaitTimeout) {
		return errors.Errorf("Wait master join failed")
	}
	if err != nil {
		return err
	}
	log.Info(constant.MasterJoinSucceedReason, "Master nodes join success")
	return e.Ctx.RefreshCtxBKECluster()
}
